package com.chukun.netty.grpc;

import com.chukun.netty.proto.*;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.util.Iterator;
import java.util.concurrent.TimeUnit;

/**
 * @author chukun
 * grpc客户端
 */
public class GrpcClient {

    public static void main(String[] args) throws InterruptedException {
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost",8899).usePlaintext().build();
        //不支持客户端流式数据的传输
        StudentServiceGrpc.StudentServiceBlockingStub blockingStub = StudentServiceGrpc.newBlockingStub(managedChannel);
        //支持异步，流式的响应式数据传输
        StudentServiceGrpc.StudentServiceStub stub = StudentServiceGrpc.newStub(managedChannel);
        try {
            StudentResponse response = blockingStub.getRealNameByUserName(StudentRequest.newBuilder().setUsername("zhangsan").build());
            System.out.println(response.getRealname());

            System.out.println("-----------------------------------------------");
            //服务端流式数据返回
            Iterator<StudentResponse> students = blockingStub.getStudentsByAge(StudentAge.newBuilder().setAge(12).build());
            while (students.hasNext()){
                StudentResponse next = students.next();
                System.out.println(next.getRealname());
            }
            System.out.println("client stream request start...........");
            //异步模式
            StreamObserver<StudentResponseList> reponseList = new StreamObserver<StudentResponseList>() {
                @Override
                public void onNext(StudentResponseList studentResponseList) {
                    studentResponseList.getStudentsList().forEach(p->{
                        System.out.println(p.getAge());
                        System.out.println(p.getCity());
                        System.out.println("********************");
                    });
                }

                @Override
                public void onError(Throwable throwable) {
                    System.out.println(throwable.getMessage());
                }

                @Override
                public void onCompleted() {
                    System.out.println("client streams transport completed.....");
                }
            };
            StreamObserver<StudentAge> studentAges = stub.getStudentWrapperByAges(reponseList);
            studentAges.onNext(StudentAge.newBuilder().setAge(10).build());
            studentAges.onNext(StudentAge.newBuilder().setAge(20).build());
            studentAges.onNext(StudentAge.newBuilder().setAge(30).build());
            studentAges.onCompleted();

            System.out.println("client-server both sides stream  start...........");
            StreamObserver<StreamResponseInfo> reponseInfos = new StreamObserver<StreamResponseInfo>() {
                @Override
                public void onNext(StreamResponseInfo streamResponseInfo) {
                    System.out.println("received :"+streamResponseInfo.getResponseInfo());
                }

                @Override
                public void onError(Throwable throwable) {
                    System.out.println(throwable.getMessage());
                }

                @Override
                public void onCompleted() {
                    System.out.println("client-server both sides stream completed");
                }
            };
            StreamObserver<StreamRequestInfo> streamRequestInfos = stub.getStudentWrapperResponseWrapper(reponseInfos);
            for (int i = 0; i < 10; i++) {
                streamRequestInfos.onNext(StreamRequestInfo.newBuilder().setRequestInfo("time_"+i).build());
            }
           streamRequestInfos.onCompleted();
        }finally {
            managedChannel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
        }
    }
}
